#include "CheckPointState.hpp"
#include "../OrderParameters.hpp"

using namespace BFT::order::statemanagement;

CheckPointState::CheckPointState(int _clients)
{
	baseDelay = 1000;
	marking = false;
	baseSeqNo = 0;
	orderedRequestCache.resize(_clients);
	for (int i = 0; i<_clients; i++)
		orderedRequestCache.at(i).reset(new OrderedEntry());
	retransDelay = std::vector<uint32_t>(_clients, baseDelay);
	currentSeqNo = 0;
	history.reset(new BFT::messages::HistoryDigest());
	execCPToken.reset();
	stableDigest.reset();
	committed=false;
	currentTime=0;
	//fprintf(stderr,"+++ Created a CheckPointState from another checkpointstate. orderedRequestCache.size() : %d\n", (int)orderedRequestCache.size());	
}

CheckPointState::CheckPointState(boost::shared_ptr<char> _bytes, int _bytes_size)
{


	baseDelay = 10;
	marking = false;
	committed = false;

	int offset = 0;	
   	
   	//read the history
	boost::shared_ptr<char> history_bytes((char*)malloc(DIGEST_SIZE), free);
	memcpy( (void*)(history_bytes.get()) , (const void*)(_bytes.get()+offset) , DIGEST_SIZE);
	offset += DIGEST_SIZE;
	history = BFT::messages::HistoryDigest::fromBytes(history_bytes, DIGEST_SIZE);
	
	//read the current time
	currentTime = ntohl(*(uint32_t*)(_bytes.get()+offset));
	offset += TIME_SIZE;
	
	//read the sequence number
	currentSeqNo = ntohl(*(uint32_t*)(_bytes.get()+offset));
	offset += SEQUENCE_NUMBER_SIZE;
	
	//read the base sequence number
	baseSeqNo = ntohl(*(uint32_t*)(_bytes.get()+offset));
	offset += SEQUENCE_NUMBER_SIZE;
	
	if (baseSeqNo % BFT::order::Parameters::checkPointInterval != 0)
	{
		fprintf(stderr,"invalid base for a checkpoint. base: %d, interval: %d\n",baseSeqNo, BFT::order::Parameters::checkPointInterval);
		exit(73);
	}
	
	//read the number of entries
	uint16_t size = ntohs( *(uint16_t*)(_bytes.get()+offset));
	offset += ENTRY_COUNT_SIZE;
	
	retransDelay = std::vector<uint32_t>(size,baseDelay);
	orderedRequestCache.resize(size);
	
	//read the entries
	uint32_t temp_reqId;
	uint32_t temp_seqNo;
	
	for (int i=0; i<size; i++)
	{
		temp_reqId = ntohl( *(uint32_t*)(_bytes.get()+offset)); 
		offset += REQUEST_ID_SIZE;
		temp_seqNo = ntohl( *(uint32_t*)(_bytes.get()+offset));
		offset += SEQUENCE_NUMBER_SIZE;
		orderedRequestCache.at(i).reset(new OrderedEntry(temp_reqId, temp_seqNo));
	}
	
	//read the execCPToken
	token_size = _bytes_size - offset;
	execCPToken.reset((char*)malloc(size),free);
	memcpy( (void*)(execCPToken.get()) , (const void*)(_bytes.get()+offset) , _bytes_size - offset );
	offset = _bytes_size;
	
}

CheckPointState::CheckPointState(boost::shared_ptr<char> _bytes, int _bytes_size, int _offset)
{

	baseDelay = 10;
	marking = false;
	committed = false;

	int offset = _offset;	
   	
   	//read the history
	boost::shared_ptr<char> history_bytes((char*)malloc(DIGEST_SIZE), free);
	memcpy( (void*)(history_bytes.get()) , (const void*)(_bytes.get()+offset) , DIGEST_SIZE);
	offset += DIGEST_SIZE;
	history = BFT::messages::HistoryDigest::fromBytes(history_bytes, DIGEST_SIZE);
	
	//read the current time
	currentTime = ntohl(*(uint32_t*)(_bytes.get()+offset));
	offset += TIME_SIZE;
	
	//read the sequence number
	currentSeqNo = ntohl(*(uint32_t*)(_bytes.get()+offset));
	offset += SEQUENCE_NUMBER_SIZE;
	
	//read the base sequence number
	baseSeqNo = ntohl(*(uint32_t*)(_bytes.get()+offset));
	offset += SEQUENCE_NUMBER_SIZE;
	
	if (baseSeqNo % BFT::order::Parameters::checkPointInterval != 0)
	{
		fprintf(stderr,"invalid base for a checkpoint. base: %d, interval: %d\n",baseSeqNo, BFT::order::Parameters::checkPointInterval);
		exit(73);
	}
	
	//read the number of entries
	uint16_t size = ntohs( *(uint16_t*)(_bytes.get()+offset));
	offset += ENTRY_COUNT_SIZE;
	
	retransDelay = std::vector<uint32_t>(size,baseDelay);
	orderedRequestCache.resize(size);
	
	//read the entries
	uint32_t temp_reqId;
	uint32_t temp_seqNo;
	
	for (int i=0; i<size; i++)
	{
		temp_reqId = ntohl( *(uint32_t*)(_bytes.get()+offset)); 
		offset += REQUEST_ID_SIZE;
		temp_seqNo = ntohl( *(uint32_t*)(_bytes.get()+offset));
		offset += SEQUENCE_NUMBER_SIZE;
		orderedRequestCache.at(i).reset(new OrderedEntry(temp_reqId, temp_seqNo));
	}
	
	//read the execCPToken
	token_size = _bytes_size - offset;
	execCPToken.reset((char*)malloc(size),free);
	memcpy( (void*)(execCPToken.get()) , (const void*)(_bytes.get()+offset) , _bytes_size - offset );
	offset = _bytes_size;
}

CheckPointState::CheckPointState(boost::shared_ptr<CheckPointState> _cps)
{
	baseDelay = 10;
	marking = false;
	committed = false;
	
	baseSeqNo = _cps.get()->getCurrentSequenceNumber();

	if (baseSeqNo % BFT::order::Parameters::checkPointInterval != 0)
	{
		fprintf(stderr,"invalid base for a checkpoint. base: %d, interval: %d\n",baseSeqNo, BFT::order::Parameters::checkPointInterval);
		exit(73);
	}
	
	currentSeqNo = _cps.get()->getCurrentSequenceNumber();
	history = _cps.get()->getHistory();
	currentTime = _cps.get()->getCurrentTime();
	orderedRequestCache.resize(_cps.get()->getEntryCount());
	retransDelay = std::vector<uint32_t>(orderedRequestCache.size(),0);
	for(unsigned int i=0; i<orderedRequestCache.size(); i++ )
	{
		orderedRequestCache.at(i) = _cps.get()->getEntry(i);
		retransDelay.at(i) = _cps.get()->getRetransDelay(i);
	}
	
	//fprintf(stderr,"+++ Created a CheckPointState from another checkpointstate. orderedRequestCache.size() : %d\n", (int)orderedRequestCache.size());	
}

boost::shared_ptr<char> CheckPointState::getBytes(int &_bytes_size)
{
	//fprintf(stderr, "getting bytes...\n");
	int bytes_size = DIGEST_SIZE //history digest
			 + TIME_SIZE //current time
			 + SEQUENCE_NUMBER_SIZE //current sequence number
			 + SEQUENCE_NUMBER_SIZE //base sequence number
			 + ENTRY_COUNT_SIZE //number of entries == number of clients
			 + orderedRequestCache.size()*(REQUEST_ID_SIZE + SEQUENCE_NUMBER_SIZE) // all the cached entries
			 + token_size; //size of the token
	int offset = 0;
	boost::shared_ptr<char> bytes ( (char*)malloc(bytes_size),free );
	
	//write the history
	boost::shared_ptr<char> temp_history = history.get()->getBytes();
	memcpy( (void*)(bytes.get()) , (const void*)(temp_history.get()) , DIGEST_SIZE);
	offset += DIGEST_SIZE;
	
	// write the current time
	*(uint32_t*)(bytes.get()+offset) = htonl(currentTime);
	offset += TIME_SIZE;
	
	//write the sequence number
	*(uint32_t*)(bytes.get()+offset) = htonl(currentSeqNo);
	offset += SEQUENCE_NUMBER_SIZE;
	
	//write the base sequence number
	*(uint32_t*)(bytes.get()+offset) = htonl(baseSeqNo);
	offset += SEQUENCE_NUMBER_SIZE;	
	
	//write the number of entries
	uint16_t size = orderedRequestCache.size();
	*(uint16_t*)(bytes.get()+offset) = htons(size);
	offset += ENTRY_COUNT_SIZE;	
	
	//write all the entries
	for (int i=0; i < size ; i++)
	{
		*(uint32_t*)(bytes.get()+offset) = htonl(orderedRequestCache.at(i).get()->getReqId());
		offset += REQUEST_ID_SIZE;	
		*(uint32_t*)(bytes.get()+offset) = htonl(orderedRequestCache.at(i).get()->getSeqNo());
		offset += SEQUENCE_NUMBER_SIZE;	
	}
	
	//write the token
	memcpy( (void*)(bytes.get()+offset) , (const void*)(execCPToken.get()) , token_size);
	offset += token_size;
	
	_bytes_size = bytes_size;
	return bytes;
}

int CheckPointState::getSize()
{
	if (!isStable())
	{
		fprintf(stderr,"cannot get the size if the check point state is not stable\n");
		exit(75);
	}
	int size;
	getBytes(size);
	return size;
}

uint32_t CheckPointState::computeSize()
{
	return DIGEST_SIZE //history digest
		 + TIME_SIZE //current time
		 + SEQUENCE_NUMBER_SIZE //current sequence number
		 + SEQUENCE_NUMBER_SIZE //base sequence number
		 + ENTRY_COUNT_SIZE //number of entries == number of clients
		 + orderedRequestCache.size()*(REQUEST_ID_SIZE + SEQUENCE_NUMBER_SIZE) // all the cached entries
		 + token_size; //size of the token
}

void CheckPointState::setCurrentTime(uint32_t _current_time)
{
	currentTime = _current_time;
}

void CheckPointState::addExecCPToken(boost::shared_ptr<char> _cp, int _cp_size, uint32_t _seqNo)
{
	if(execCPToken.use_count() != 0)
	{
		fprintf(stderr,"already have a checkpoint\n");
		exit(74);
	}
	if (_seqNo % BFT::order::Parameters::checkPointInterval != 0)
	{
		fprintf(stderr,"not a checkpoint interval: %d\n",_seqNo);
		exit(73);
	}
	
	execCPToken = _cp;
	token_size = _cp_size;
}

boost::shared_ptr<char> CheckPointState::getExecCPToken(int& _size)
{
	_size = token_size;
	return execCPToken;
}

void CheckPointState::addNextBatch(boost::shared_ptr<BFT::messages::CommandBatch> _cb, uint32_t _seqNo,
						  boost::shared_ptr<BFT::messages::HistoryDigest> _hd)
{
	if (_seqNo != currentSeqNo)
	{
		fprintf(stderr,"invalid batch update. currentseqno: %d, _seqno: %d\n",currentSeqNo, _seqNo);
		exit(76);
	}
	if (_seqNo >= currentSeqNo + BFT::order::Parameters::checkPointInterval)
	{
		fprintf(stderr,"invalid addition of a batch. currentseqno: %d , _seqno: %d\n",currentSeqNo,_seqNo);
		exit(77);
	}
	
	boost::shared_ptr<std::vector<boost::shared_ptr<BFT::messages::Entry> > > entries = _cb.get()->getEntries();
	uint16_t entry_count = _cb.get()->getEntryCount();
	
	int tempclient;
	for (int i=0; i< entry_count ; i++)
	{
		tempclient = entries.get()->at(i).get()->getClient();
		if( orderedRequestCache.at(tempclient).get()->getSeqNo() < baseSeqNo )
		{
			orderedRequestCache.at(tempclient).reset( 
				new OrderedEntry( entries.get()->at(i).get()->getRequestId(), _seqNo) );
		}
		else
		{
			orderedRequestCache.at(tempclient).get()->set(entries.get()->at(i).get()->getRequestId() , _seqNo);
		}
		
		retransDelay.at(tempclient) = baseDelay;
	}
	
	history = _hd;
	currentSeqNo++;
}

void CheckPointState::addNextBatch(boost::shared_ptr<BFT::messages::CertificateEntry> _cert, uint32_t _seqNo)
{
	addNextBatch(_cert.get()->getCommandBatch(), _seqNo, _cert.get()->getHistoryDigest());
}

void CheckPointState::addNextBatch(boost::shared_ptr<BFT::messages::NextBatch> _nb)
{
	addNextBatch(_nb.get()->getCommands(), _nb.get()->getSeqNo(), _nb.get()->getHistory());
}

boost::shared_ptr<BFT::messages::Digest> CheckPointState::getDigest() 
{
	int temp_bytes_size;
	boost::shared_ptr<char> temp_bytes = getBytes(temp_bytes_size);
	boost::shared_ptr<BFT::messages::Digest> temp_digest (new BFT::messages::Digest(temp_bytes, temp_bytes_size));
	return temp_digest;
}

void CheckPointState::markingStable()
{
	if (marking)
	{
		fprintf(stderr,"already marking stable\n");
		exit(77);
	}
	marking = true;
}

void CheckPointState::printInfo()
{
	fprintf(stdout,"CheckPointState information:\n");
	fprintf(stdout,"base seqno: %d\ncurrent seqno: %d\n",baseSeqNo, currentSeqNo);
	fprintf(stdout,"printing the %d entries in orderedRequestCache and retransDelay\n", (int)orderedRequestCache.size());
	for(unsigned int i=0 ; i<orderedRequestCache.size();i++ )
	{
		fprintf(stdout,"entry %d:\n",i);
		orderedRequestCache.at(i).get()->printInfo();
		fprintf(stdout,"retransDelay.at(i): %d\n",retransDelay.at(i));
	}
	//history.get()->printInfo();
	fprintf(stdout,"execCPToken: ");
	for (int i=0; i< token_size; i++)
		//fprintf(stdout,"|%02X",(unsigned char)buffer[i]);
		fprintf(stdout,"|%02X",(unsigned char)execCPToken.get()[i]);
	fprintf(stdout,"\n");
	fprintf(stdout,"currentTime: %d\n",currentTime);
	fprintf(stdout,"committed: %d\n", committed?1:0 );
	fprintf(stdout,"marking: %d\n", marking?1:0 );
	fprintf(stdout,"end CheckPointState information.\n");	
}

void CheckPointState::makeStable()
{
//	fprintf(stderr, "before acquiring the lock...\n");	
	boost::unique_lock<boost::recursive_mutex> lock(condition_mutex);
//	fprintf(stderr, "lock acquired...\n");	
	if (!marking)
	{
		fprintf(stderr, "should be marking stable before making stable\n");
		exit(78);
	}
//	else
//	{
//		fprintf(stderr, "not marking, going on...\n");	
//	}
	
	if (isStable())
	{
		fprintf(stderr,"should not be making stable if it is already stable\n");
		exit(79);
	}
//	else
//	{
//		fprintf(stderr, "not stable, going on...\n");	
//	}
	
	if (currentSeqNo % BFT::order::Parameters::checkPointInterval != 0)
	{
	  std::cerr << "Curr SN: " << currentSeqNo << std::endl;
		fprintf(stderr,"must be at a checkpoint interval to be made stable\n");
		//exit(80);
	}
//	else
//	{
//		fprintf(stderr, "in a cp interval, going on...\n");	
//	}
	
	if(hasExecCP())
	{
//		fprintf(stderr, "hasExecCP...\n");	
		int temp_bytes_size;
//		fprintf(stderr, "hasExecCP 2...\n");	
		boost::shared_ptr<char> temp_bytes = getBytes(temp_bytes_size);
//		fprintf(stderr, "hasExecCP 3...\n");	
		stableDigest.reset( new BFT::messages::Digest(temp_bytes, temp_bytes_size));
//		fprintf(stderr, "hasExecCP 4...\n");	
	}
	else
	{
		fprintf(stderr,"attempting to make a cp stable without a cp token\n");
		exit(81);
	}
//	fprintf(stderr, "before notification...\n");	
	stable_condition.notify_all();
//	fprintf(stderr, "after notification, returning...\n");	
}

void CheckPointState::commit()
{	
	boost::unique_lock<boost::recursive_mutex> lock(condition_mutex);
	
	while(!isStable())
	{
		stable_condition.wait(lock);
	}
	
	committed = true;
}

bool CheckPointState::isStable()
{
	boost::unique_lock<boost::recursive_mutex> lock(condition_mutex);
	return stableDigest.use_count() != 0;	
}

bool CheckPointState::isCommitted()
{
	boost::unique_lock<boost::recursive_mutex> lock(condition_mutex);
	if (committed && !isStable())
	{
		fprintf(stderr,"cannot be commited without being stable\n");
		exit(82);
	}
	
	return committed;
}

boost::shared_ptr<BFT::messages::Digest> CheckPointState::getStableDigest()
{
	boost::unique_lock<boost::recursive_mutex> lock(condition_mutex);
	return stableDigest;
}
